package cn.easy.rpc.io;

import cn.easy.rpc.config.ConfigHolder;
import cn.easy.rpc.core.ServiceRegistrar;
import cn.easy.rpc.core.ServiceSchema;
import cn.easy.rpc.trans.TransModel;
import lombok.AllArgsConstructor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Arrays;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @author zwb
 * @create 2020-09-19 22:53
 **/
@Component
public class RpcServer implements ApplicationListener<ContextRefreshedEvent>, ApplicationContextAware {

    private static final Log logger = LogFactory.getLog(RpcServer.class);

    private static final ThreadPoolExecutor worker = new ThreadPoolExecutor(3, 8, 15L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(32), r -> {
        Thread thread = new Thread(r);
        thread.setName("RpcServer worker");
        return thread;
    });

    private ApplicationContext applicationContext = null;
    private ServiceRegistrar serviceRegistrar = ConfigHolder.config;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        CompletableFuture.runAsync(() -> {
            ServerSocket serverSocket = null;
            try {
                serverSocket = new ServerSocket(serviceRegistrar.getPort());
            } catch (IOException e) {
                e.printStackTrace();
            }
            do {
                Socket accept = null;
                try {
                    if ((accept = serverSocket.accept()) != null) {
                        logger.info("support a rpc request " + accept.getInetAddress());
                        worker.submit(new RpcClientHandler(accept));
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            } while (true);
        });
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @AllArgsConstructor
    class RpcClientHandler implements Runnable {
        private final Socket socket;

        @Override
        public void run() {
            ObjectInputStream is = null;
            ObjectOutputStream os = null;
            try {
                is = new ObjectInputStream(new BufferedInputStream(
                        socket.getInputStream()));
                TransModel transModel = (TransModel) is.readObject();
                os = new ObjectOutputStream(socket.getOutputStream());
                ServiceSchema serviceSchema = serviceRegistrar.getServices().stream().filter(schema -> schema.getSchema().equals(transModel.getSchema())).findFirst().orElse(null);
                Object target = applicationContext.getBean(serviceSchema.getBeanName());
                Class<?>[] paramType = new Class<?>[transModel.getArgs().length];
                for (int i = 0; i < transModel.getArgs().length; i++) {
                    paramType[i] = transModel.getArgs()[i].getClass();
                }
                Method method = target.getClass().getMethod(transModel.getMethodName(), paramType);
                os.writeObject(method.invoke(target, transModel.getArgs()));
            } catch (Exception ex) {
                ex.printStackTrace();
            } finally {
                try {
                    if (is != null)
                        is.close();
                    if (os != null)
                        os.close();
                    if (socket != null)
                        socket.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
}
